package com.netposa.service.Schedule;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.io.*;
import java.time.LocalDateTime;
import java.util.UUID;

@Component
@EnableScheduling
public class KafkaProducer {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaProducer.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 发送字符串
     */
    @Scheduled(cron = "0/1 * * * * ?")
    public void send(){
        String message = UUID.randomUUID().toString();
        ListenableFuture future = kafkaTemplate.send("test", message);
        future.addCallback(o -> LOG.debug("send-消息发送成功：{}" , message), throwable -> LOG.error("消息发送失败：消息——{},时间——{}" ,message,LocalDateTime.now()));
    }

    /**
     * 发送byte[]
     */
//    @Scheduled(cron = "0/2 * * * * ?")
    public void send_byte(){
        String message = UUID.randomUUID().toString();
        ListenableFuture future = kafkaTemplate.send("test_byte", message.getBytes());
        future.addCallback(o -> LOG.debug("send_byte-消息发送成功：{}" , message), throwable -> LOG.error("消息发送失败：消息——{},时间——{}" , message, LocalDateTime.now()));
    }

    @Scheduled(cron = "0/2 * * * * ?")
    public void send_img(){
        String message = UUID.randomUUID().toString();
        File file = new File("C:\\Users\\admin\\Pictures\\1.jpg");
        byte[] buffer = null;
        try {
            FileInputStream fis = new FileInputStream(file);
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            byte[] b = new byte[1024];
            int n;
            while ((n = fis.read(b)) != -1)
            {
                bos.write(b, 0, n);
            }
            fis.close();
            bos.close();
            buffer = bos.toByteArray();

        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }

        ListenableFuture future = kafkaTemplate.send("test_img", buffer);
        future.addCallback(o -> LOG.debug("send-消息发送成功：{}" , message), throwable -> LOG.error("消息发送失败：消息——{},时间——{}" ,message,LocalDateTime.now()));
    }
}
